package cn.echcz.register

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable
import scala.concurrent.duration._

class Register extends Actor {
  val HEARTBEAT = "heartbeat"
  val REGISTER_SUCCESS = "register_success"
  val CHECK = "check"
  // 设置检查间隔为20秒
  val CHECK_INTERVAL = 20000

  // 服务列表的排序是以health的大小为主的
  implicit val ord : Ordering[ServerInfo] = new Ordering[ServerInfo]{
    override def compare(x: ServerInfo, y: ServerInfo): Int = {
      if (x.health == y.health){
        (x.id).compareTo(y.id)
      } else {
        y.health - x.health
      }
    }
  }

  private[this] val servers = mutable.TreeSet[ServerInfo]()
  private[this] val serverHeartbeat = mutable.HashSet[String]()
  private[this] val clients = mutable.HashSet[ClientInfo]()
  private[this] val clientHeartbeat = mutable.HashSet[(String, String)]()

  override def preStart(): Unit = {
    // 定时向自己发送需要检查消息
    import context.dispatcher
    context.system.scheduler.schedule(0 millis, CHECK_INTERVAL millis,self, CHECK)
  }

  override def receive: Receive = {
    // 接收服务提供者注册
    case server: ServerInfo => {
      servers += server
      sender ! REGISTER_SUCCESS
    }
    // 接收服务消费者注册
    case client: ClientInfo => {
      clients += client
      sender ! REGISTER_SUCCESS
    }
    // 接收服务提供者心跳
    case (HEARTBEAT, id: String) => {
      serverHeartbeat += id
    }
    // 接收服务消费者心跳
    case (HEARTBEAT, host: String, port: String) => {
      clientHeartbeat += Tuple2(host, port)
    }
    // 当接收到自己的定时检查消息时，清理服务列表并传送给客户端
    case CHECK => {
      // 清理服务端列表
      servers.retain(x => serverHeartbeat.contains(x.id))
      // 清空服务心跳库
      serverHeartbeat.retain(p => false)
      // 清理客户端列表
      clients.retain(x => clientHeartbeat.contains((x.host, x.port)))
      // 清理客户心跳库
      clientHeartbeat.retain(p => false)
      // 给每个客户端发送服务端列表
      clients.foreach(x => {
        context.actorSelection(s"akka.tcp://clientActorSystem@${x.host}:${x.port}/user/clientActor") !
          servers
      })
    }
  }
}

object Register{
  def apply(host: String, port: String): ActorRef = {
    // 设置actorSystem的配置
    val configSetting =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
       """.stripMargin
    val config = ConfigFactory.parseString(configSetting)
    // 创建ActorSystem
    val registerActorSystem = ActorSystem("registerActorSystem", config)
    registerActorSystem.actorOf(Props(new Register), "registerActor")
  }
}
